//! 集群管理器实现
//!
//! 提供完整的集群节点管理和状态协调功能，支持：
//! - 节点的动态注册与发现
//! - 心跳检测和健康状态监控
//! - 领导者选举和故障转移
//! - 集群状态的实时监控和告警
//! - 基于事件驱动的状态变更通知
//!
//! # 核心概念
//! - **集群配置**：定义集群的基本参数和行为策略
//! - **节点信息**：包含节点的身份、状态、权重等关键信息
//! - **集群事件**：节点加入、离开、故障等状态变更事件
//! - **集群状态**：运行、降级、停止等整体健康状态
//!
//! # 使用示例
//! ```rust
//! use nacos_cluster::{ClusterManager, ClusterConfig, ClusterNodeInfo};
//! use std::sync::Arc;
//!
//! #[tokio::main]
//! async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
//!     let config = ClusterConfig {
//!         cluster_name: "my-cluster".to_string(),
//!         heartbeat_timeout: 30,
//!         election_timeout: 3000,
//!         current_node_id: "node-1".to_string(),
//!         initial_nodes: vec!["node-1:8848".to_string(), "node-2:8848".to_string()],
//!     };
//!     
//!     let manager = Arc::new(ClusterManager::new(config));
//!     let current_node = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
//!     
//!     manager.initialize(current_node).await?;
//!     
//!     Ok(())
//! }
//! ```

use chrono::Utc;
use crate::node::{ClusterNodeInfo, NodeHealth};
use nacos_core::{NodeState, Result, NacosError};
use tokio::sync::RwLock;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time;
use tracing::{error, info, warn};

/// 集群配置
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterConfig {
    /// 集群名称
    pub cluster_name: String,
    /// 心跳超时时间（秒）
    pub heartbeat_timeout: i64,
    /// 选举超时时间（毫秒）
    pub election_timeout: u64,
    /// 当前节点ID
    pub current_node_id: String,
    /// 初始节点列表
    pub initial_nodes: Vec<String>,
}

impl Default for ClusterConfig {
    fn default() -> Self {
        Self {
            cluster_name: "nacos-cluster".to_string(),
            heartbeat_timeout: 30,
            election_timeout: 3000,
            current_node_id: String::new(),
            initial_nodes: Vec::new(),
        }
    }
}

/// 集群事件
#[derive(Debug, Clone)]
pub enum ClusterEvent {
    /// 节点加入
    NodeJoined(ClusterNodeInfo),
    /// 节点离开
    NodeLeft(String),
    /// 节点故障
    NodeFailed(String),
    /// 领导者变更
    LeaderChanged(Option<String>),
    /// 集群状态变更
    ClusterStateChanged,
}

/// 集群状态
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ClusterState {
    /// 初始化
    Initializing,
    /// 运行中
    Running,
    /// 降级模式（少数节点）
    Degraded,
    /// 停止
    Stopped,
}

/// 集群管理器
#[derive(Debug)]
pub struct ClusterManager {
    /// 集群配置
    config: ClusterConfig,
    /// 节点信息
    nodes: Arc<RwLock<HashMap<String, ClusterNodeInfo>>>,
    /// 当前领导者
    leader_id: Arc<RwLock<Option<String>>>,
    /// 集群状态
    state: Arc<RwLock<ClusterState>>,
    /// 事件发送器
    event_sender: Arc<RwLock<Option<mpsc::UnboundedSender<ClusterEvent>>>>,
    /// 当前节点信息
    current_node: Arc<RwLock<Option<ClusterNodeInfo>>>,
}

impl ClusterManager {
    /// 创建新的集群管理器
    pub fn new(config: ClusterConfig) -> Self {
        info!("Creating new cluster manager for cluster: {}", config.cluster_name);
        
        Self {
            config,
            nodes: Arc::new(RwLock::new(HashMap::new())),
            leader_id: Arc::new(RwLock::new(None)),
            state: Arc::new(RwLock::new(ClusterState::Initializing)),
            event_sender: Arc::new(RwLock::new(None)),
            current_node: Arc::new(RwLock::new(None)),
        }
    }

    /// 初始化集群管理器
    pub async fn initialize(&self, current_node: ClusterNodeInfo) -> Result<()> {
        info!("Initializing cluster manager for node: {}", current_node.node_id);
        
        // 设置当前节点
        let mut current = self.current_node.write().await;
        *current = Some(current_node.clone());
        drop(current);

        // 添加当前节点到集群
        self.add_node(current_node.clone()).await?;
        
        // 设置集群状态为运行中
        let mut state = self.state.write().await;
        *state = ClusterState::Running;
        drop(state);

        // 触发节点加入事件
        self.send_event(ClusterEvent::NodeJoined(current_node)).await;
        
        info!("Cluster manager initialized successfully");
        Ok(())
    }

    /// 设置事件发送器
    pub async fn set_event_sender(&self, sender: mpsc::UnboundedSender<ClusterEvent>) {
        let mut event_sender = self.event_sender.write().await;
        *event_sender = Some(sender);
    }

    /// 添加节点
    pub async fn add_node(&self, node: ClusterNodeInfo) -> Result<()> {
        let node_id = node.node_id.clone();
        
        {
            let mut nodes = self.nodes.write().await;
            if nodes.contains_key(&node_id) {
                warn!("Node {} already exists in cluster", node_id);
                return Ok(());
            }
            
            nodes.insert(node_id.clone(), node.clone());
        }
        
        info!("Node {} added to cluster", node_id);
        self.send_event(ClusterEvent::NodeJoined(node)).await;
        
        Ok(())
    }

    /// 移除节点
    pub async fn remove_node(&self, node_id: &str) -> Result<()> {
        let node = {
            let mut nodes = self.nodes.write().await;
            nodes.remove(node_id)
        };

        if let Some(node) = node {
            info!("Node {} removed from cluster", node_id);
            self.send_event(ClusterEvent::NodeLeft(node_id.to_string())).await;
            
            // 如果是领导者，触发重新选举
            if node.is_leader() {
                self.trigger_election().await?;
            }
            
            Ok(())
        } else {
            warn!("Node {} not found in cluster", node_id);
            Err(NacosError::NotFound(node_id.to_string()))
        }
    }

    /// 更新节点状态
    pub async fn update_node_state(&self, node_id: &str, state: NodeState) -> Result<()> {
        let mut nodes = self.nodes.write().await;
        if let Some(node) = nodes.get_mut(node_id) {
            let old_state = node.state.clone();
            node.state = state.clone();
            
            if old_state != state {
                info!("Node {} state changed from {:?} to {:?}", node_id, old_state, state);
                
                match state {
                    NodeState::Failed => {
                        self.send_event(ClusterEvent::NodeFailed(node_id.to_string())).await;
                        if node.is_leader() {
                            self.trigger_election().await?;
                        }
                    }
                    NodeState::Stopped | NodeState::Left => {
                        self.send_event(ClusterEvent::NodeLeft(node_id.to_string())).await;
                        if node.is_leader() {
                            self.trigger_election().await?;
                        }
                    }
                    _ => {}
                }
            }
            
            Ok(())
        } else {
            Err(NacosError::NotFound(node_id.to_string()))
        }
    }

    /// 更新节点心跳
    pub async fn update_node_heartbeat(&self, node_id: &str) -> Result<()> {
        let mut nodes = self.nodes.write().await;
        if let Some(node) = nodes.get_mut(node_id) {
            node.update_heartbeat();
            node.health = NodeHealth::Healthy;
            Ok(())
        } else {
            Err(NacosError::NotFound(node_id.to_string()))
        }
    }

    /// 获取所有节点
    pub async fn get_all_nodes(&self) -> Vec<ClusterNodeInfo> {
        let nodes = self.nodes.read().await;
        nodes.values().cloned().collect()
    }

    /// 获取健康节点
    pub async fn get_healthy_nodes(&self) -> Vec<ClusterNodeInfo> {
        let nodes = self.nodes.read().await;
        nodes
            .values()
            .filter(|node| node.is_healthy(self.config.heartbeat_timeout))
            .cloned()
            .collect()
    }

    /// 获取节点信息
    pub async fn get_node(&self, node_id: &str) -> Option<ClusterNodeInfo> {
        let nodes = self.nodes.read().await;
        nodes.get(node_id).cloned()
    }

    /// 获取当前领导者
    pub async fn get_leader(&self) -> Option<ClusterNodeInfo> {
        let leader_id = self.leader_id.read().await;
        if let Some(id) = leader_id.as_ref() {
            self.get_node(id).await
        } else {
            None
        }
    }

    /// 设置领导者
    pub async fn set_leader(&self, node_id: &str) -> Result<()> {
        let mut nodes = self.nodes.write().await;
        
        // 清除之前的领导者
        for node in nodes.values_mut() {
            node.set_as_follower();
        }
        
        // 设置新的领导者
        if let Some(node) = nodes.get_mut(node_id) {
            node.set_as_leader();
            
            let mut leader = self.leader_id.write().await;
            *leader = Some(node_id.to_string());
            
            info!("New leader elected: {}", node_id);
            self.send_event(ClusterEvent::LeaderChanged(Some(node_id.to_string()))).await;
            
            Ok(())
        } else {
            Err(NacosError::NotFound(node_id.to_string()))
        }
    }

    /// 触发选举
    pub async fn trigger_election(&self) -> Result<()> {
        info!("Triggering leader election");
        
        let healthy_nodes = self.get_healthy_nodes().await;
        if healthy_nodes.is_empty() {
            warn!("No healthy nodes available for election");
            return Ok(());
        }
        
        // 简单的选举算法：选择权重最高的节点
        let new_leader = healthy_nodes
            .iter()
            .max_by_key(|node| node.weight)
            .unwrap();
        
        self.set_leader(&new_leader.node_id).await
    }

    /// 检查集群健康状态
    pub async fn check_cluster_health(&self) -> Result<()> {
        let healthy_nodes = self.get_healthy_nodes().await;
        let total_nodes = self.get_all_nodes().await.len();
        
        let new_state = if healthy_nodes.is_empty() {
            ClusterState::Stopped
        } else if healthy_nodes.len() < total_nodes / 2 + 1 {
            ClusterState::Degraded
        } else {
            ClusterState::Running
        };
        
        let mut state = self.state.write().await;
        if *state != new_state {
            *state = new_state;
            info!("Cluster state changed to: {:?}", new_state);
            self.send_event(ClusterEvent::ClusterStateChanged).await;
        }
        
        Ok(())
    }

    /// 获取集群状态
    pub async fn get_cluster_state(&self) -> ClusterState {
        *self.state.read().await
    }

    /// 获取当前节点
    pub async fn get_current_node(&self) -> Option<ClusterNodeInfo> {
        let current = self.current_node.read().await;
        current.clone()
    }

    /// 获取集群大小
    pub async fn get_cluster_size(&self) -> usize {
        let nodes = self.nodes.read().await;
        nodes.len()
    }

    /// 获取集群统计信息
    pub async fn get_cluster_stats(&self) -> ClusterStats {
        let nodes = self.nodes.read().await;
        let healthy_nodes: Vec<_> = nodes
            .values()
            .filter(|node| node.is_healthy(self.config.heartbeat_timeout))
            .collect();
        
        let leader_id = self.leader_id.read().await.clone();
        let state_guard = self.state.read().await;
        
        ClusterStats {
            total_nodes: nodes.len(),
            healthy_nodes: healthy_nodes.len(),
            leader_id,
            cluster_state: *state_guard,
        }
    }

    /// 发送事件
    async fn send_event(&self, event: ClusterEvent) {
        let sender = self.event_sender.read().await;
        if let Some(sender) = sender.as_ref() {
            if let Err(e) = sender.send(event) {
                error!("Failed to send cluster event: {}", e);
            }
        }
    }

    /// 启动心跳检查
    pub async fn start_heartbeat_check(self: Arc<Self>) {
        let nodes = self.nodes.clone();
        let timeout = self.config.heartbeat_timeout;
        let manager = self.clone();
        
        tokio::spawn(async move {
            let mut interval = time::interval(Duration::from_secs(5));
            
            loop {
                interval.tick().await;
                
                let nodes_snapshot = {
                    let nodes = nodes.read().await;
                    nodes.clone()
                };
                
                for (node_id, node) in nodes_snapshot {
                    if !node.is_healthy(timeout) {
                        warn!("Node {} appears to be unhealthy", node_id);
                        let _ = manager.update_node_state(&node_id, NodeState::Failed).await;
                    }
                }
                
                let _ = manager.check_cluster_health().await;
            }
        });
    }

    /// 关闭集群管理器
    pub async fn shutdown(&self) -> Result<()> {
        info!("Shutting down cluster manager");
        
        let mut state = self.state.write().await;
        *state = ClusterState::Stopped;
        
        self.send_event(ClusterEvent::ClusterStateChanged).await;
        
        Ok(())
    }
}

impl Clone for ClusterManager {
    fn clone(&self) -> Self {
        Self {
            config: self.config.clone(),
            nodes: self.nodes.clone(),
            leader_id: self.leader_id.clone(),
            state: self.state.clone(),
            event_sender: self.event_sender.clone(),
            current_node: self.current_node.clone(),
        }
    }
}


/// 集群统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClusterStats {
    /// 总节点数
    pub total_nodes: usize,
    /// 健康节点数
    pub healthy_nodes: usize,
    /// 领导者ID
    pub leader_id: Option<String>,
    /// 集群状态
    pub cluster_state: ClusterState,
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::sync::mpsc;

    #[tokio::test]
    async fn test_cluster_manager() {
        let config = ClusterConfig {
            cluster_name: "test-cluster".to_string(),
            current_node_id: "node1".to_string(),
            ..Default::default()
        };
        
        let manager = Arc::new(ClusterManager::new(config));
        
        // 创建测试节点
        let node1 = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
        let node2 = ClusterNodeInfo::new("127.0.0.1", 8849, 2);
        
        // 初始化
        manager.initialize(node1.clone()).await.unwrap();
        
        // 添加节点
        manager.add_node(node2.clone()).await.unwrap();
        
        // 验证节点数量
        assert_eq!(manager.get_cluster_size(), 2);
        
        // 测试领导者选举
        manager.set_leader(&node1.node_id).await.unwrap();
        assert_eq!(manager.get_leader().unwrap().node_id, node1.node_id);
        
        // 测试节点移除
        manager.remove_node(&node2.node_id).await.unwrap();
        assert_eq!(manager.get_cluster_size(), 1);
        
        // 测试事件系统
        let (tx, mut rx) = mpsc::unbounded_channel();
        manager.set_event_sender(tx);
        
        let node3 = ClusterNodeInfo::new("127.0.0.1", 8850, 1);
        manager.add_node(node3.clone()).await.unwrap();
        
        let event = rx.recv().await.unwrap();
        match event {
            ClusterEvent::NodeJoined(node) => assert_eq!(node.node_id, node3.node_id),
            _ => panic!("Expected NodeJoined event"),
        }
    }

    #[tokio::test]
    async fn test_cluster_health_check() {
        let config = ClusterConfig {
            heartbeat_timeout: 1, // 1秒超时用于测试
            ..Default::default()
        };
        
        let manager = Arc::new(ClusterManager::new(config));
        let node = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
        
        manager.initialize(node.clone()).await.unwrap();
        
        // 初始状态应该是健康的
        assert_eq!(manager.get_cluster_state(), ClusterState::Running);
        
        // 模拟心跳超时
        let mut node_mut = manager.nodes.write();
        if let Some(node) = node_mut.get_mut(&node.node_id) {
            node.last_heartbeat = Utc::now() - chrono::Duration::seconds(2);
        }
        drop(node_mut);
        
        // 手动触发健康检查
        manager.check_cluster_health().await.unwrap();
        
        // 应该检测到节点不健康
        let stats = manager.get_cluster_stats();
        assert_eq!(stats.healthy_nodes, 0);
    }

    #[tokio::test]
    async fn test_leader_election_on_node_failure() {
        let config = ClusterConfig {
            heartbeat_timeout: 1,
            ..Default::default()
        };
        
        let manager = Arc::new(ClusterManager::new(config));
        let node1 = ClusterNodeInfo::new("127.0.0.1", 8848, 1);
        let node2 = ClusterNodeInfo::new("127.0.0.1", 8849, 2);
        
        manager.initialize(node1.clone()).await.unwrap();
        manager.add_node(node2.clone()).await.unwrap();
        
        // 设置node1为领导者
        manager.set_leader(&node1.node_id).await.unwrap();
        assert_eq!(manager.get_leader().unwrap().node_id, node1.node_id);
        
        // 模拟领导者故障
        manager.update_node_state(&node1.node_id, NodeState::Failed).await.unwrap();
        
        // 应该触发新的选举，node2成为新的领导者
        assert_eq!(manager.get_leader().unwrap().node_id, node2.node_id);
    }
}